package com.song.hbase01;

import java.io.IOException;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.HTable;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.util.Bytes;

public class Hbase001 {

	/**
	 * 获取制定的行的列
	 */
	public static void putData()
	{

		Configuration conf = HBaseConfiguration.create();
		try {
			HTable htable = new HTable(conf, "example");
			for(int i = 0 ;i<100 ;i++)
			{
				Put put = new Put(Bytes.toBytes("row"+i));
				put.add(Bytes.toBytes("family1"),Bytes.toBytes("qulifier1"), Bytes.toBytes("value1"));
				htable.put(put);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
	}
	
	public static void getData()
	{
		Configuration conf = HBaseConfiguration.create();
		try {
			HTable htable = new HTable(conf, "table2");
			Get get = new Get(Bytes.toBytes("row2"));
			get.getMaxVersions();
			Result r = htable.get(get);
			byte[] bs = r.getValue(Bytes.toBytes("cf"), Bytes.toBytes("attrib1"));
			String val = new String(bs);
			System.out.println(val);
			
			List<KeyValue> kvs = r.getColumn( Bytes.toBytes("cf"), Bytes.toBytes("attrib1") );
			for(KeyValue kv :kvs)
			{
				System.out.println(new String(kv.getKey()) +" :  "+new String(kv.getValue()));
			}
			
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	/**
	 *   遍历hbase 的表
	 */
	public static void scanTable()
	{
		Configuration conf = HBaseConfiguration.create();
		try {
			HTable htable = new HTable(conf, "table2");
			Scan scan = new Scan();
			scan.addColumn(Bytes.toBytes("cf"), Bytes.toBytes("attrib1"));
			scan.setStartRow(Bytes.toBytes("row1"));
			scan.setStopRow(Bytes.toBytes("row50"));
			ResultScanner rscan = htable.getScanner(scan);
			
			for(Result rs =rscan.next(); rs != null; rs = rscan.next())
			{
				byte[] bs = rs.getValue(Bytes.toBytes("cf"), Bytes.toBytes("attrib1"));
				String val = new String(bs);
				System.out.println(val);
			}
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}
	
	public static void scankeyvalTable(String tablename)
	{
		
		
		Configuration conf = HBaseConfiguration.create();
		
		try {
			HTable htable = new HTable(conf, Bytes.toBytes(tablename));
			Scan scan = new Scan();
			
			ResultScanner rscan = htable.getScanner(scan);
			
			for( Result rs = rscan.next() ; rs!=null; rs  =rscan.next())
			{
				List<KeyValue>  keyvals = rs.list();
				for(KeyValue kv :keyvals)
				{
					System.out.println( new String(kv.getFamily())+":"+new String(kv.getQualifier()) +"  "+ new String(kv.getValue()) );;
				}
				
			}
			
			
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		
	}
	
	
	
	
	public static void main(String[] args) {
//		putData();
//		getData();
//		scanTable();
		scankeyvalTable("data_char_7_6");
	}
	
	
	
}
